package main

/**
该程序是为了解决rtmp流分布式访问而写的代理转发程序，rtmp推流可以通过负载均衡的方式进行多台服务器处理，但在
用户访问时，需要知道当前流所分配的服务器，而RTMP协议中流的访问地址是包含在流的内部，所以必须解析流内容后，才可
以获得流的访问地址，所以为了解决RTMP集群化的问题，特编写输出流代理服务，通过该服务来处理多个服务器的流的输出；

version:20220119
author：xiaobo
date:2022-01-19
**/

import (
	"flag"
	"fmt"
	"net"
	"os"
	"strconv"
	"time"

	"github.com/garyburd/redigo/redis"
	"github.com/rs/zerolog"
	"gopkg.in/ini.v1"
	"yunshicloud.com/gproxy/core"
)

var pool *redis.Pool
var call_url = ""
var qtype = 0 //查询服务器的类型，0：http;1:redis

var logger = zerolog.New(os.Stdout).With().Timestamp().Logger()

func main() {

	conf := flag.String("c", "./config.ini", "配置文件地址")
	help := flag.Bool("help", false, "print usage")
	bind := flag.String("bind", "0.0.0.0:1935", "The address to bind to")
	debuglevel := flag.Bool("debug", false, "default:disable")
	// backend := flag.String("backend", "10.0.2.9:1935", "The backend server address")
	flag.Parse()

	zerolog.SetGlobalLevel(zerolog.InfoLevel)
	if *debuglevel {
		zerolog.SetGlobalLevel(zerolog.DebugLevel)
	}
	if *help {
		flag.Usage()
		return
	}
	if *debuglevel {
		//use default bind
		logger.Info().Str("debug", "disable").Msg("debug")
	}

	if *bind == "" {
		//use default bind
		logger.Info().Str("bind", *bind).Msg("use default bind")
	}

	if *conf == "" {
		//use default bind
		logger.Info().Str("conf", *conf).Msg("use default conf")
	}
	//获取配置文件的访问链接，通过http的方式获取实际流所在的服务器；
	call_url = InitConfigFromIni(*conf)
	if len(call_url) == 0 && qtype == 0 {

		logger.Error().Msg("获取配置文件内容错误!")
		os.Exit(1)
	}
	success, err := RunProxy(*bind)
	if !success {
		logger.Error().Err(err).Send()
		os.Exit(1)
	}
}

//接收rtmp的连接请求，并进行代理转发;
func RunProxy(bind string) (bool, error) {
	listener, err := net.Listen("tcp", bind)
	if err != nil {
		return false, err
	}
	defer listener.Close()
	//通过chan来限制最大的并发数量，单机最大为1024个，如果超过1024则进行等待;
	ch := make(chan int, 1024)
	for {
		socket, err := listener.Accept()
		if err != nil {
			logger.Error().Err(err).Send()
		} else {
			ch <- 1
			conn := core.NewConn(socket, 1024*4)
			go core.ConnectionHandler(conn, call_url, pool, qtype, ch)
		}
	}
}

//初始化读取配置文件内容;[http]url:
func InitConfigFromIni(conf string) string {

	_, err := os.Stat(conf)
	if err != nil {
		logger.Error().Err(err).Msg("读取配置文件错误")
		os.Exit(1)
	}
	cfg, err := ini.Load(conf)
	if err != nil {
		fmt.Printf("Fail to read file: %v", err)
		os.Exit(1)
	}
	url := cfg.Section("http").Key("url").String()
	if url != "" {
		logger.Debug().Str("url:", call_url).Msg("获取http_getserver的访问链接")
		return url
	}

	constr := cfg.Section("redis").Key("conn").String()
	dbstr := cfg.Section("redis").Key("db").String()
	passwd := cfg.Section("redis").Key("passwd").String()
	poolstr := cfg.Section("redis").Key("pool").String()
	if constr != "" {
		qtype = 1
		initRedis(constr, dbstr, passwd, poolstr) //生成redis的连接池
		logger.Debug().Str("conn:", constr).Msg("建立redis连接池")
	}
	return ""
}

func initRedis(constr, dbstr, passwd, poolstr string) {

	db := 0
	poolnum := 0
	if dbstr != "" {
		db, _ = strconv.Atoi(dbstr)
	}
	if poolstr != "" {
		poolnum, _ = strconv.Atoi(poolstr)
	}
	setdb := redis.DialDatabase(db) // 设置db
	setPasswd := redis.DialPassword(passwd)
	pool = &redis.Pool{
		MaxIdle:     4,
		MaxActive:   poolnum,
		IdleTimeout: 100,
		Dial: func() (redis.Conn, error) {
			c, err := redis.Dial("tcp", constr, setdb, setPasswd)
			if err != nil {
				return nil, err
			}
			return c, err
		},
		TestOnBorrow: func(c redis.Conn, t time.Time) error {
			_, err := c.Do("PING")
			return err
		},
	}
}
